package bigdata.hermesfuxi.eagle.etl.utils;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;

/**
 * 自定义Kafka反序列化器
 * 
 * 该类实现了KafkaDeserializationSchema接口，用于自定义Kafka消息的反序列化逻辑。
 * 主要功能是将Kafka消息的元数据（topic、partition、offset）和消息内容组合成Tuple2格式。
 * 
 * 主要特点：
 * 1. 提取Kafka消息的元数据信息（topic、partition、offset）
 * 2. 将元数据组合成全局唯一ID
 * 3. 将消息内容转换为UTF-8字符串
 * 4. 返回Tuple2<全局ID, 消息内容>格式的数据
 * 
 * 使用场景：
 * - 需要跟踪消息来源的场景
 * - 需要保证消息唯一性的场景
 * - 需要根据消息来源进行不同处理的场景
 * 
 * @author hermesfuxi
 * @version 1.0
 * @since 2021
 */
public class MyKafkaDeserializationSchema implements KafkaDeserializationSchema<Tuple2<String, String>> {

    /**
     * 判断是否为流的结束
     * 
     * 由于Kafka是持续的数据流，该方法始终返回false，
     * 表示流不会结束。
     * 
     * @param nextElement 下一个元素
     * @return 始终返回false
     */
    @Override
    public boolean isEndOfStream(Tuple2<String, String> nextElement) {
        return false;
    }

    /**
     * 反序列化Kafka消息
     * 
     * 该方法将Kafka的ConsumerRecord反序列化为Tuple2格式，
     * 包含全局唯一ID和消息内容。
     * 
     * 全局ID格式：topic-partition-offset
     * 例如：user_behavior-0-12345
     * 
     * @param record Kafka消费者记录
     * @return Tuple2<全局ID, 消息内容>
     * @throws Exception 反序列化过程中的异常
     */
    @Override
    public Tuple2<String, String> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        // 提取Kafka消息元数据
        String topic = record.topic();
        int partition = record.partition();
        long offset = record.offset();
        
        // 将消息内容转换为UTF-8字符串
        String value = new String(record.value(), StandardCharsets.UTF_8);
        
        // 组合全局唯一ID并返回Tuple2
        return Tuple2.of(topic + "-" + partition + "-" + offset, value);
    }

    /**
     * 获取输出类型信息
     * 
     * 返回反序列化器输出的数据类型信息，用于Flink的类型推断。
     * 
     * @return Tuple2<String, String>的类型信息
     */
    @Override
    public TypeInformation<Tuple2<String, String>> getProducedType() {
        return TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
        });
    }
}
